package com.kangjia.config;

import java.util.concurrent.ThreadPoolExecutor;

import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;

import com.rabbitmq.client.ConnectionFactory;

@Configuration
public class RabbitmqConfig {

	@Value("${rabbitmq.host}")
	private String host;
	@Value("${rabbitmq.username}")
	private String username;
	@Value("${rabbitmq.password}")
	private String password;
	@Value("${rabbitmq.port}")
	private int port;
	@Value("${rabbitmq.vhost}")
	private String vhost;

	@Bean
	public ConnectionFactory rabbitmqConnectionFactory() {
		ConnectionFactory factory = new ConnectionFactory();
		factory.setHost(host);
		factory.setUsername(username);
		factory.setPassword(password);
		factory.setPort(port);
		factory.setConnectionTimeout(5000);
		factory.setVirtualHost(vhost);
		return factory;
	}

	@Bean
	public ThreadPoolTaskExecutor rabbitThreadPoolTaskExecutor() {
		ThreadPoolTaskExecutor tpte = new ThreadPoolTaskExecutor();
		// 核心线程数
		tpte.setCorePoolSize(3);
		// 最大线程数
		tpte.setMaxPoolSize(20);
		// 队列最大长度 >=mainExecutor.maxSize
		tpte.setQueueCapacity(1000);
		// 线程池维护线程所允许的空闲时间
		tpte.setKeepAliveSeconds(300);
		ThreadPoolExecutor.CallerRunsPolicy crp = new ThreadPoolExecutor.CallerRunsPolicy();
		// 线程池对拒绝任务(无线程可用)的处理策略
		tpte.setRejectedExecutionHandler(crp);
		return tpte;
	}

	@Bean
	public CachingConnectionFactory cachingConnectionFactory(ConnectionFactory factory, ThreadPoolTaskExecutor tpte) {
		CachingConnectionFactory caching = new CachingConnectionFactory(factory);
		caching.setChannelCacheSize(25);
		caching.setExecutor(tpte);
		return caching;
	}
	
	@Bean
	public RabbitTemplate amqpTemplate(CachingConnectionFactory ccf) {
		RabbitTemplate rt = new RabbitTemplate();
		rt.setConnectionFactory(ccf);
		rt.setReplyTimeout(1000);
		rt.setExchange("kj.report");
		return rt;
	}
}
